package com.nokia.ion.sdu.cdr_decoder;

import java.io.IOException;
import java.sql.*;
import java.util.concurrent.LinkedBlockingQueue;

/**
 * Created by vicwang on 2017/9/21.
 */
public class MysqlWriter extends CdrWriter {

    private static final String driver = "com.mysql.cj.jdbc.Driver";

    private static final String url = "jdbc:mysql://%s:%d/%s";

    private static final String CREATE_TABLE = "CREATE TABLE `cdr` (\n" +
            "  `node_id` varchar(32) DEFAULT NULL,\n" +
            "  `apn` varchar(64) DEFAULT NULL,\n" +
            "  `p_addr` varchar(32) DEFAULT NULL,\n" +
            "  `s_addr` varchar(32) DEFAULT NULL,\n" +
            "  `msisdn` bigint(20) DEFAULT NULL,\n" +
            "  `imei` bigint(20) DEFAULT NULL,\n" +
            "  `imsi` bigint(20) DEFAULT NULL,\n" +
            "  `rat_type` tinyint(4) NOT NULL,\n" +
            "  `uli` varchar(64) DEFAULT NULL,\n" +
            "  `open_time` timestamp NOT NULL,\n" +
            "  `duration` int(10) UNSIGNED NOT NULL,\n" +
            "  `cause` tinyint(4) NOT NULL,\n" +
            "  `charging_id` int(10) UNSIGNED NOT NULL\n" +
            ") ENGINE=InnoDB DEFAULT CHARSET=utf8mb4;";

    private static final String SQL = "insert into cdr(node_id, s_addr, imsi, rat_type, uli, open_time, duration, cause, charging_id) values (?, ?, ?, ?, ?, ?, ?, ?, ?)";

    private Statement statement;
    private Connection conn;
    private PreparedStatement prest;

    private Configuration.MysqlConfigure mysqlConfigure;

    private int idx = 0;
    private int commitMax = 10000;

    public MysqlWriter(LinkedBlockingQueue<Cdr> queue, Configuration configuration) throws IOException, SQLException, ClassNotFoundException {
        super(queue, configuration);

        mysqlConfigure = configuration.mysql;

        commitMax = mysqlConfigure.commitBufSize;

        Class.forName(driver);
        conn = DriverManager.getConnection(String.format(url, mysqlConfigure.host, mysqlConfigure.port, mysqlConfigure.dbName),
                mysqlConfigure.username, mysqlConfigure.password);
        if (!conn.isClosed())
            System.out.println("Succeeded connecting to the Mysql!");
        else {
            throw new SQLException("Failed connecting to Mysql server!");
        }

        conn.setAutoCommit(false);
        statement = conn.createStatement();
        statement.executeUpdate("drop table if EXISTS cdr");
        statement.executeUpdate(CREATE_TABLE);
        conn.commit();

        prest = conn.prepareStatement(SQL);
    }

    @Override
    public void write(Cdr cdr) {
        try {
            prest.setString(1, cdr.nodeID);
            prest.setString(2, cdr.servingNodeAddress);
            prest.setLong(3, cdr.servedIMSI);
            prest.setInt(4, cdr.rATType);
            prest.setString(5, cdr.userLocationInformation);
            prest.setDate(6, new Date(cdr.recordOpeningTime.getTime()));
            prest.setInt(7, cdr.duration);
            prest.setInt(8, cdr.causeForRecClosing);
            prest.setInt(9, cdr.chargingId);
            prest.addBatch();

            idx++;
            if (idx >= commitMax) {
                prest.executeBatch();
                conn.commit();
                idx = 0;
            }
        } catch (SQLException e) {
            System.err.println(e);
        }
    }

    @Override
    void close() {
        try {
            conn.close();
        } catch (SQLException e) {
            System.err.println(e);
        }
    }
}
